package com.gopay.common.zk.leader.support;

import java.io.Closeable;
import java.io.IOException;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @author kz-tu
 *
 */
public abstract class AbstractLeaderSelectorListener implements LeaderSelectorListener, Closeable, Runnable {
	
	protected final Logger logger = LoggerFactory.getLogger(getClass());

	private CuratorFramework client;
	private LeaderSelector leaderSelector;
	
	private final byte[] mutex = new byte[0];
	
	private volatile boolean isConnected = true;
	
	private String leaderNodePath;
	private String connectString;
	private int baseSleepTimeMs;
	private int maxRetries;

	public void setLeaderNodePath(String leaderNodePath) {
		this.leaderNodePath = leaderNodePath;
	}

	public void setConnectString(String connectString) {
		this.connectString = connectString;
	}

	public void setBaseSleepTimeMs(int baseSleepTimeMs) {
		this.baseSleepTimeMs = baseSleepTimeMs;
	}

	public void setMaxRetries(int maxRetries) {
		this.maxRetries = maxRetries;
	}

	@Override
	public void stateChanged(CuratorFramework client, ConnectionState newState) {
		logger.info("[stateChanged]newState={}", newState);
		if ((newState == ConnectionState.SUSPENDED)
				|| (newState == ConnectionState.LOST)) {
			isConnected = false;
			synchronized (mutex) {
				mutex.notify();
			}
		} else if ((newState == ConnectionState.CONNECTED)
				|| (newState == ConnectionState.RECONNECTED)
				|| (newState == ConnectionState.READ_ONLY)) {
			isConnected = true;
		}
	}

	@Override
	public void takeLeadership(CuratorFramework client) throws Exception {
		logger.info("[takeLeadership]take leadership.");
		
		processBiz();
		
		synchronized (mutex) {
			while (isConnected) {
				mutex.wait();
				logger.info("[takeLeadership]relinquishing leadership.");
			}			
		}
		
		relinquishLeadership();
	}
	
	/**
	 * 获得leadership后，进行相应的业务逻辑处理 
	 */
	public abstract void processBiz() throws Exception;
	
	/**
	 * 放弃leadership后，进行相应的业务逻辑处理 
	 */
	public abstract void relinquishLeadership() throws Exception;
	
	@Override
	public void close() throws IOException {
		CloseableUtils.closeQuietly(client);
		CloseableUtils.closeQuietly(leaderSelector);
	}

	@Override
	public void run() {
		try {
			client = CuratorFrameworkFactory.newClient(this.connectString, new ExponentialBackoffRetry(this.baseSleepTimeMs, this.maxRetries));
			client.start();

			leaderSelector = new LeaderSelector(client, getLeaderNodePath(), this);
			leaderSelector.autoRequeue();
			leaderSelector.start();
			
			synchronized (this) {
				// 设置当前线程一直为等待状态，除非当前线程中断。
				while(true){
					wait();
				}
			}
		}catch (InterruptedException ie) {
			logger.error(Thread.currentThread().getName() + " was interrupted...");
            Thread.currentThread().interrupt();
		} finally {
			try {
				close();
			} catch (IOException e) {
				e.printStackTrace();
				logger.error("Close SelectorListener Error", e);
			}
		}
	}

	public boolean isConnected() {
		return isConnected;
	}

	public String getLeaderNodePath() {
		return leaderNodePath;
	}

	public String getConnectString() {
		return connectString;
	}

	public int getBaseSleepTimeMs() {
		return baseSleepTimeMs;
	}

	public int getMaxRetries() {
		return maxRetries;
	}
}
